package com.yifeng.repo.flink.data.transport.streaming.connectors.mysql;

import java.util.List;

import org.antlr.v4.runtime.CharStreams;
import org.antlr.v4.runtime.CommonTokenStream;
import org.antlr.v4.runtime.tree.ParseTreeWalker;
import org.apache.commons.collections.CollectionUtils;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.util.Collector;
import org.apache.kafka.connect.data.Schema;
import org.apache.kafka.connect.data.Struct;
import org.apache.kafka.connect.source.SourceRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.ververica.cdc.debezium.DebeziumDeserializationSchema;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import com.yifeng.repo.flink.data.transport.config.DdlConstants;
import com.yifeng.repo.flink.data.transport.dto.DbTypeEnums;
import com.yifeng.repo.flink.data.transport.dto.DdlContent;
import com.yifeng.repo.flink.data.transport.dto.DdlDataPacket;
import com.yifeng.repo.flink.data.transport.dto.HistoryRecord;
import com.yifeng.repo.flink.data.transport.streaming.connectors.mysql.antlr.listener.MysqlDdlParserListener;
import com.yifeng.repo.flink.data.transport.utils.ddl.DdlUtils;

import io.debezium.ddl.parser.mysql.generated.MySqlLexer;
import io.debezium.ddl.parser.mysql.generated.MySqlParser;

/**
 * 自定义序列化器
 */
public class MysqlDebeziumDeserializationSchema implements DebeziumDeserializationSchema<String> {
	
	private static  final Logger logger = LoggerFactory.getLogger(MysqlDebeziumDeserializationSchema.class);
	
	private static final long serialVersionUID = 1L;
	
    private String database;
    
    private boolean schemaChangeToDownstream;
	
	public MysqlDebeziumDeserializationSchema(boolean schemaChangeToDownstream,String database) {
		this.database = database;
		this.schemaChangeToDownstream = schemaChangeToDownstream;
	}
	
    @Override
    public void deserialize(SourceRecord sourceRecord, Collector<String> collector)
        throws Exception {
        //判断是否是DDL变更数据
    	Schema keySchema = sourceRecord.keySchema();
    	if(keySchema != null && DdlConstants.MYSQL_SCHEMA_CHANGE_EVENT_KEY_NAME.equalsIgnoreCase(keySchema.name())) {
    		if(!schemaChangeToDownstream) {
    			return;
    		}
    		dealDDLData(sourceRecord,collector);
    		return;
    	}
    	dealDMlData(sourceRecord,collector);
    }
    
    /**
     * 处理DML数据
     * @param sourceRecord 数据记录
     * @param collector 结果集合
     * @throws Exception 
     */
    private void dealDMlData(SourceRecord sourceRecord, Collector<String> collector) throws Exception {
    	JsonDebeziumDeserializationSchema schema = new JsonDebeziumDeserializationSchema();
    	schema.deserialize(sourceRecord, collector);
    }
    
    /**
     * 处理DDL数据
     * @param sourceRecord 数据记录
     * @param collector 结果集合
     */
    private void dealDDLData(SourceRecord sourceRecord, Collector<String> collector) {
    	DdlDataPacket ddlDataPacket = new DdlDataPacket();
    	ddlDataPacket.setOp(DdlConstants.SCHEMA_CHANGE_OP_TYPE);
    	ddlDataPacket.setDbType(DbTypeEnums.MYSQL.getType());
    	ddlDataPacket.setDatabase(database);
    	Struct value = (Struct)sourceRecord.value();
    	Struct source = value.getStruct("source");
    	ddlDataPacket.setTb(String.valueOf(source.get("table")));
    	//将historyRecord转换为bean
    	HistoryRecord historyRecord = JSONObject.parseObject(JSON.parse(JSON.toJSONString(value.get("historyRecord"))).toString(), HistoryRecord.class);
    	//根据ddl语句转换数据给下游系统
    	List<DdlContent> ddlContentList = parseDdlContent(historyRecord,ddlDataPacket.getTb());
    	if(CollectionUtils.isEmpty(ddlContentList)) {
    		return;
    	}
        ddlDataPacket.setDdlContentList(ddlContentList);
    	ddlDataPacket.setTsMs(System.currentTimeMillis());
    	logger.info("DDL变更转换后的数据:{}",JSONObject.toJSONString(ddlDataPacket));
    	collector.collect(JSONObject.toJSONString(ddlDataPacket));
    }
    
    /**
     * 解析DDL语句，转换成下游需要的数据
     * @param historyRecord ddl数据
     * @param tableName 表名
     * @return ddl变更数据
     */
    private static List<DdlContent> parseDdlContent(HistoryRecord historyRecord,String tableName) {
    	String ddl = historyRecord.getDdl();
    	MySqlLexer lexer = new MySqlLexer(CharStreams.fromString(ddl.toUpperCase())); 
        MySqlParser parser = new MySqlParser(new CommonTokenStream(lexer)); // 语法分析
        MysqlDdlParserListener mysqlDdlParserListener = new MysqlDdlParserListener(tableName);
        ParseTreeWalker.DEFAULT.walk(mysqlDdlParserListener,parser.root());
    	//如果没有数据则不传递给下游
        if(mysqlDdlParserListener.getColumnMap().isEmpty() && mysqlDdlParserListener.getRenameMap().isEmpty()) {
        	return null;
        }
        return DdlUtils.parseDdlContent(mysqlDdlParserListener.getRenameMap(),mysqlDdlParserListener.getColumnMap(),historyRecord);
    }
    
    @Override
    public TypeInformation<String> getProducedType() {
        return BasicTypeInfo.STRING_TYPE_INFO;
    } 
    
}